package com.cnpc.bds.sdk.sink.kafka.meta;

import lombok.Builder;
import lombok.Data;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Properties;

/**
 * Kafka消费者配置.
 */
@Data
@Builder
public class ConsumerCfg {

    private TopicMeta topicMeta;

    /**
     * config.
     */
    private String autoOffsetReset;
    /**
     * config.
     */
    private String keyDeserializer;
    /**
     * config.
     */
    private String valueDeserializer;
    /**
     * config.
     */
    private Integer fetchMinSize;
    /**
     * config.
     */
    private Integer fetchMaxWaitMs;
    /**
     * config.
     */
    private Integer maxPollRecords;

    public Properties properties() {
        Properties config = new Properties();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, topicMeta.getBrokers());
        config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer);
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer);
        config.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, fetchMinSize);
        config.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, fetchMaxWaitMs);
        config.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);
        return config;
    }

    public static ConsumerCfgBuilder defaultCfg() {
        return ConsumerCfg.builder()
                .autoOffsetReset("latest")
                .keyDeserializer(StringDeserializer.class.getName())
                .valueDeserializer(StringDeserializer.class.getName())
                .fetchMinSize(1024 * 2)
                .fetchMaxWaitMs(1500)
                .maxPollRecords(1000);
    }


}
